package com.kingsoft.dc.khaos.module.spark.source

import java.util.Properties

import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.metadata.{Dependency, KhaosStructField}
import com.kingsoft.dc.khaos.module.spark.metadata.source.MppSourceConfig
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.DataFrame
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.parse

import scala.collection.mutable.ArrayBuffer


class MppDirectSource extends SourceStrategy with Logging{
  private var _mpp_Config: MppSourceConfig = null
  private var _kc: KhaosContext = null

  // 物理地址
  private var _host: String = null
  private var _port: String = null
  private var _username: String = null
  private var _password: String = null
  private var _instansename: String = null


  /** 数据抽取 */
  def source(kc: KhaosContext,
             module_id: String,
             config: String,
             dependence: Dependency,
             ds_config: Map[String, String]): DataFrame = {

    //根据json解析出module
    implicit val formats = DefaultFormats
    val mppConfig: MppSourceConfig = parse(config, true).extract[MppSourceConfig]
    this._mpp_Config = mppConfig
    this._kc = kc

    //权限校验
    //    MetaUtils.checkReadAuth(kc,
    //      mppConfig.db_name,
    //      mppConfig.table_name,
    //      mppConfig.extender.auth.clazz,
    //      compact(render(mppConfig.extender.auth.params)))

    // 获取物理地址
    initMetaData(ds_config: Map[String, String])

    // 通过jdbc直连方式获取dataframe
    val driver = "org.postgresql.Driver"
    val prop = new Properties
    prop.put("driver", driver)
    prop.put("user", _username)
    prop.put("password", _password)

    val gscReadOptionMap = Map(
      "url" -> getConnectUrl(),
      "user" -> _username,
      "password" -> _password,
      "dbschema" -> _mpp_Config.db_name,
      "dbtable" -> _mpp_Config.table_name,
      "partitionColumn" -> "a1")

    val mppDF = kc.sparkSession.read.format("greenplum")
      .options(gscReadOptionMap)
      .load()

    if (mppDF.first() != null) {
      log.info("读取MPP源表成功...")
    }

    // 过滤条件
    if (_mpp_Config.filter != null && !_mpp_Config.filter.trim.equals("")) {
      mppDF.filter(_mpp_Config.filter)
    } else {
      mppDF
    }
  }

  /**
   * 从数据管理获取mpp物理地址
   */
  def initMetaData(ds_config: Map[String, String]): Unit = {
    log.info("初始化物理地址...")
    //    val className = _hawq_Config.extender.meta.clazz
    //    val dbname = _hawq_Config.db_name
    //    val tblname = _hawq_Config.table_name
    //
    //    val mppConnect = MetaUtils.getMPPMeta(_kc,
    //      dbname,
    //      tblname,
    //      className,
    //      compact(render(_hawq_Config.extender.auth.params)), this).getDsMppConnect
    //
    //    _host = mppConnect.getHost
    //    _port = mppConnect.getPort
    //    _username = mppConnect.getUsername
    //    _password = mppConnect.getPassword
    //    _instansename = mppConnect.getInstanceName
    //    _dataSourceType = mppConnect.getSourceMode

    _host = ds_config("host")
    _port = ds_config("port")
    _username = ds_config("username")
    _password = ds_config("password")
    _instansename = ds_config("instansename")
  }

  /**
   * 初始化物理地址
   */
  //  def initMetaData() = {
  //    log.info("初始化物理地址...")
  //    val className = _mpp_Config.extender.meta.clazz
  //    implicit val formats = DefaultFormats
  //    val metaParamsJson: ExtendedMetaParams = _mpp_Config.extender.meta.params.extract[ExtendedMetaParams]
  //    val metaParamsStr: String =
  //      s"""
  //         |{"project_id":"${metaParamsJson.project_id}","ds_id":"${metaParamsJson.ds_id}","ds_type":"${metaParamsJson.ds_type}","ds_name":"${metaParamsJson.ds_name}"}
  //              """.stripMargin
  //    val dbname = _mpp_Config.db_name
  //    val tblname = _mpp_Config.table_name
  //
  //    val mppConnect = MetaUtils.getMPPMeta(_kc, dbname, tblname, className, metaParamsStr, this).getDsMppConnect
  //
  //    _host = mppConnect.getHost
  //    _port = mppConnect.getPort
  //    _username = mppConnect.getUsername
  //    _password = mppConnect.getPassword
  //    _instansename = mppConnect.getInstanceName
  //  }

  /**
   * 获取jdbc连接url
   *
   * @return
   */
  def getConnectUrl(): String = {
    val host = _host
    val port = _port
    val dbName = _instansename
    val url = s"jdbc:postgresql://${host}:${port}/${dbName}"
    log.info("url ==> " + url)
    url.trim
  }

  /**
   * 获取mpp内部表名
   *
   * @return 表名
   */
  def getMppSchemaAndTableName() = {
    val schema = _mpp_Config.db_name
    val mppTable = _mpp_Config.table_name
    schema + "." + mppTable.toLowerCase
  }


  /** 获取上游的Schema */
  override def schema(dc: KhaosContext,
                      config: String,
                      dependence: Dependency): List[KhaosStructField] = {
    val fieldSchema = ArrayBuffer[KhaosStructField]()
    implicit val formats = DefaultFormats
    val mppInfo = parse(config, true).extract[MppSourceConfig]
    val extrFields = mppInfo.extract_fields
    for (ef <- extrFields) {
      fieldSchema += KhaosStructField(ef.field, ef.data_type)
    }
    fieldSchema.toList
  }
}
